package com.atguigu.flink.chapter5.sink;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

import java.util.Properties;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2020/12/21 11:09
 */
public class Flink01_Sink_Kafka {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);
        Properties props = new Properties();
        props.put("bootstrap.servers", "hadoop162:9092");
        env
          .socketTextStream("hadoop162", 9999)
//          .addSink(new FlinkKafkaProducer<String>("sensor", new SimpleStringSchema(), props));
          .addSink(new FlinkKafkaProducer<String>("hadoop162:9092,hadoop163:9092", "sensor", new SimpleStringSchema()));

        env.execute();
    }
}
